package com.yytek.common.client.mqtt;

import com.yytek.common.client.base.IMqClient;
import com.yytek.common.client.base.MqConfig;
import com.yytek.common.client.base.MqMessage;
import com.yytek.common.client.base.MessageCallBack;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.UnsupportedEncodingException;
import java.util.HashSet;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;


public class PahoMqttClient implements IMqClient {
    private int connectCheckTime = 6000;
    private Timer reconnectTimer;
    private MqttClient client;
    private MqConfig mqConfig;
    private Set<String> topics;
    private MessageCallBack messageCallBack;

    public PahoMqttClient(MqConfig mqConfig) {
        topics = new HashSet<String>();
        this.mqConfig = mqConfig;
    }

    public boolean isConnected() {
        if (client != null) {
            return client.isConnected();
        } else
            return false;
    }

    public void setMessageCallBack(MessageCallBack messageCallBack) {
        this.messageCallBack = messageCallBack;
    }

    public boolean open() {
        if (client != null && client.isConnected())
            return true;
        if (mqConfig.getClientId() == null) {
            mqConfig.setClientId(MqttClient.generateClientId());
        }
        try {
            connectClient();
        } catch (MqttException e) {
            e.printStackTrace();
            return false;
        }
        reconnectTimer = new Timer("MQTT Reconnect: " + mqConfig.getClientId());
        reconnectTimer.scheduleAtFixedRate(new ReconnectTask(), 0, connectCheckTime);
        return true;
    }

    private void connectClient() throws MqttException, MqttSecurityException {
        MemoryPersistence persistence = new MemoryPersistence();
        client = new MqttClient(mqConfig.getBroker(), mqConfig.getClientId(), persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
		/*
		会话清除标识session
		值为false时，broker存储该客户端订阅的主题及未接收的消息(QoS大于1) 客户端重启后将重新接收消息；
		值为true时 broker不存储以上数据，客户端重启也不会接收到离线消息
		*/
        connOpts.setCleanSession(true);
        //连接丢失时客户端是否自动尝试重新连接到服务器
        connOpts.setAutomaticReconnect(false);
        //会话心跳时间，单位为秒。即多长时间确认一次Client端是否在线  服务器会每隔1.5*60秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        connOpts.setKeepAliveInterval(60);
        //允许同时发送几条消息( 未收到broker确认信息，MaxInflight+1，收到确认信息  MaxInflight-1，)
        //此参数也可以理解为  未收到broker确认信息 数量 。当qos=1时，数值超过设定上限，mqtt报错
        connOpts.setMaxInflight(100);
        if (!mqConfig.getUserName().isEmpty() && !mqConfig.getPassword().isEmpty()) {
            connOpts.setUserName(mqConfig.getUserName());
            connOpts.setPassword(mqConfig.getPassword().toCharArray());
        }
        //超时时间
        client.setTimeToWait(mqConfig.getTimeToWait());
        client.connect(connOpts);
    }

    private class ReconnectTask extends TimerTask {
        public void run() {
            if (!client.isConnected()) {
                try {
                    connectClient();
                    if (client.isConnected()) {
                        for (String topic : topics) {
                            subscribeTopic(topic);
                        }
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }

        }
    }


    private boolean subscribeTopic(String destination) {
        try {
            client.subscribe(destination, new IMqttMessageListener() {
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    try {
                        if (messageCallBack != null) {
                            messageCallBack.onPublish(topic, new MqMessage(message.getId(), message.getPayload()), destination);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
            return true;
        } catch (MqttException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean subscribe(String topic) {
        topics.add(topic);
        if (client.isConnected()) {
            subscribeTopic(topic);
        }

        return true;
    }


    public boolean unsubscribe(String destination) {
        topics.remove(destination);
        try {
            client.unsubscribe(destination);
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return true;
    }

    public synchronized boolean publish(String topics, byte[] content, int qos, int timout) {
        MqttMessage message;
        try {
            if (client.isConnected()) {
//				message = new MqttMessage(content);
//				message.setQos(qos);
//				MqttTopic tocpic = client.getTopic(topics);
//				MqttDeliveryToken tocket = tocpic.publish(message);
//				tocket.waitForCompletion(timout);
//				return true;
                client.getTopic(topics).publish(content, qos, false);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    public boolean publish(String topic, String content, int qos, int timout) {
        try {
            return publish(topic, content.getBytes("UTF-8"), qos, timout);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean publish(String topic, String content) {
        return publish(topic, content, 0, 1000);
    }

    public boolean publish(String topic, MqMessage content) {
        return publish(topic, content.getPayload(), content.getQos(), 1000);
    }


    public void close() {
        try {
            reconnectTimer.cancel();
            if (client != null)
                client.disconnect();
        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }


    public int getConnectCheckTime() {
        return connectCheckTime;
    }

    public void setConnectCheckTime(int connectCheckTime) {
        this.connectCheckTime = connectCheckTime;
    }

    public Timer getReconnectTimer() {
        return reconnectTimer;
    }

    public void setReconnectTimer(Timer reconnectTimer) {
        this.reconnectTimer = reconnectTimer;
    }

    public MqttClient getClient() {
        return client;
    }

    public void setClient(MqttClient client) {
        this.client = client;
    }
}
